package com.xy.springboot_3.reactor.demo;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * Reactor 框架说明演示
 */
@Slf4j
public class ReactorDemo {

    @SneakyThrows
    public static void main(String[] args) {

        new ReactorDemo().asyncThread();
        //TimeUnit.SECONDS.sleep(1L);
    }

    public void asyncThread(){

        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4,true);

        final Flux<String> flux = Flux
                .range(1, 2)
                .map(i -> 10 + i)
                .log()
                .publishOn(s)
                .map(i -> "value " + i)
                ;

        //只要不指定线程池，默认发布者用的线程就是订阅者的线程；
        new Thread(() -> flux.subscribe(System.out::println),"Thread-nickName").start();
    }


    // 异步演示
    @SneakyThrows
    public void async(){

        Flux.range(1,10)
                .publishOn(Schedulers.boundedElastic())  // 发布者使用的 异步线程池
                .map(item -> {
                    log.error(Thread.currentThread().getName());
                    return item+10;
                })
                .subscribeOn(Schedulers.boundedElastic())  // 订阅者使用的异步线程池
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        log.info(Thread.currentThread().getName());
                        request(1L);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info(Thread.currentThread().getName());
                        request(1L);
                    }
                });

        Thread.sleep(10000L);

       // Schedulers.immediate(); // 使用当前订阅者线程。 不存在线程池调度。

      //  Schedulers.boundedElastic(); // 使用默认提供的调度器  CPU * 10 , 队列 1000

     //   Schedulers.fromExecutor(new ThreadPoolExecutor(6,10,60,TimeUnit.SECONDS,new LinkedBlockingQueue<>(1000))); // 自定义线程池调度器。


    }

    public void handle(){
        Flux.range(1,10)
                .log()
                // handle 核心用法 就是将其 数据结构进行变更 然后再通过 skin 进行传递
                .handle((v,skin)->{
                    // 更改数据结构
                    skin.next(v +"---");
                })
                .log()
                .subscribe();
    }

    public void create(){
        Flux.create((fluxSink -> {
        })).log().subscribe();
    }



    public void generate(){
        LongAdder longAdder = new LongAdder();
        Flux.generate((synchronousSink) ->{
                    if (longAdder.longValue() >= 10L) {
                        synchronousSink.complete();
                    }
            synchronousSink.next(longAdder.intValue());
            longAdder.increment();

        })
                .log()
                .subscribe();
    }

    public void limit(){

        Flux.range(1,1000)
                .log()
                .limitRate(100)
                .subscribe();
    }

    // 缓冲区，意义在于，我能一次性 多拿数据
    public  void buffer(){
        /**
         * [1, 2, 3]
         * [4, 5, 6]
         * [7, 8, 9]
         * [10]
         */
        Flux.range(1,10)
                .buffer(3)
                .doOnNext(v->{
                    log.info(v.getClass().toGenericString());
                })
                .log()
                .doOnEach(System.out::println)
                .subscribe(System.out::println);

    }

    public void demo1(){
        Flux.range(1,100)
                .log()
                .doOnNext(System.out::println)
                .subscribe();
    }

    // 钩子函数
    public void hook(){
        Flux.range(1,10)
                .doOnComplete(()->log.info("doOnComplete 执行了"))
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                       log.info("hookOnSubscribe");
                       request(1L);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info(String.valueOf(value));
                        request(1L);

                    }

                    @Override
                    protected void hookOnComplete() {
                        log.info("hookOnComplete");
                    }

                    @Override
                    protected void hookOnError(Throwable throwable) {
                        super.hookOnError(throwable);
                    }

                    @Override
                    protected void hookOnCancel() {
                        super.hookOnCancel();
                    }

                    @Override
                    protected void hookFinally(SignalType type) {
                        super.hookFinally(type);
                    }
                })
                ;

    }

    //消费者调用 cancle() 取消流的订阅；
    public void cancel(){
        Flux<String> flux = Flux.range(1, 10)
                .map(i -> {
                    System.out.println("map..."+i);
                    if(i==9) {
                        i = 10/(9-i); //数学运算异常;  doOnXxx
                    }
                    return "哈哈：" + i;
                }); //流错误的时候，把错误吃掉，转为正常信号


//        flux.subscribe(); //流被订阅; 默认订阅；
//        flux.subscribe(v-> System.out.println("v = " + v));//指定订阅规则： 正常消费者：只消费正常元素


//        flux.subscribe(
//                v-> System.out.println("v = " + v), //流元素消费
//                throwable -> System.out.println("throwable = " + throwable), //感知异常结束
//                ()-> System.out.println("流结束了...") //感知正常结束
//        );


        // 流的生命周期钩子可以传播给订阅者。
        //  a() {
        //      data = b();
        //  }
        flux.subscribe(new BaseSubscriber<String>() {

            // 生命周期钩子1： 订阅关系绑定的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                // 流被订阅的时候触发
                System.out.println("绑定了..."+subscription);

                //找发布者要数据
                request(1); //要1个数据
//                requestUnbounded(); //要无限数据
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("数据到达，正在处理："+value);
                if(value.equals("哈哈：5")){
                    cancel(); //取消流
                }
                request(1); //要1个数据
            }


            //  hookOnComplete、hookOnError 二选一执行
            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束...");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常..."+throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消...");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调...一定会被执行");
            }
        });
    }

    //自定义流的信号感知回调
    public void process(){
        Flux<Integer> flux = Flux.range(1, 10);
        flux.subscribe(
                v-> System.out.println("v = " + v), //流元素消费
                throwable -> System.out.println("throwable = " + throwable), //感知异常结束
                ()-> System.out.println("流结束了...") //感知正常结束
        );
    }
}
